package ca.tatham.scheduler;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Observable;

public class Scheduler extends Observable {
	private final TaskQueue<ScheduledTask> m_taskQueue;
	private final Map<Long, ScheduledTask> m_tasks;
	private final ResourceList m_activeResources;
	private final Map<Object, Resource> m_resources;

	private boolean m_schedulingStarted = false;

	private static Scheduler instance;

	public static Scheduler instance() {
		if (instance == null) {
			instance = new Scheduler();
		}
		return instance;
	}

	private Scheduler() {
		m_taskQueue = new TaskQueue<ScheduledTask>();
		m_tasks = new HashMap<Long, ScheduledTask>();
		m_activeResources = new ResourceList();
		m_resources = new HashMap<Object, Resource>();
	}

	public static void schedule(Task task) {
		instance.scheduleTask(task);
	}

	private void scheduleTask(Task task) {
		m_schedulingStarted = true;
		Log.log(this, "Scheduling " + task.getName());
		final ScheduledTask sTask = new ScheduledTask(task);
		m_tasks.put(sTask.getId(), sTask);
		ScheduledTask currentTask = m_taskQueue.peek();
		m_taskQueue.add(sTask);
		if (m_taskQueue.peek() == sTask) {
			boolean released = false;
			if (currentTask != null) {
				if (Thread.currentThread().getId() == currentTask.getId()) {
					new ContextSwitcher(sTask).start();
					released = true;
				}
				currentTask.block();
			}

			if (!released) {
				sTask.release();
			}
		}
	}

	/**
	 * All tasks must be registered before any can be scheduled. This allows us
	 * to know the priority ceiling of each resource before beginning.
	 * 
	 * @param task
	 */
	public void registerTask(Task task) {
		if (m_schedulingStarted) {
			throw new IllegalStateException(
					"Cannot register tasks after scheduling has begun!");
		}
		for (Object object : task.getUsedResources()) {
			Resource resource = m_resources.get(object);
			if (resource == null) {
				m_resources.put(object, new Resource(object, task
						.getOriginalPriority()));
			} else if (task.getOriginalPriority() < resource.getCeiling()) {
				resource.setCeiling(task.getOriginalPriority());
			}
		}
	}

	public void checkHighestPriorityTask() {
		ScheduledTask task = m_taskQueue.popComplete();
		if (task == null) {
			allTasksComplete();
		} else {
			task.release();
		}
	}

	private void allTasksComplete() {
		setChanged();
		notifyObservers();
	}

	private int getCurrentPriorityCeiling() {
		Resource peek = m_activeResources.peek();
		if (peek == null) {
			return Integer.MAX_VALUE;
		}
		return peek.getCeiling();
	}

	public static void lock(Object object) throws SchedulerException {
		instance.lockResource(object);
	}

	private void lockResource(Object object) throws SchedulerException {
		Log.log(this, "Request to lock " + object);
		ScheduledTask task = m_tasks.get(Thread.currentThread().getId());
		Resource resource = m_activeResources.getResource(object);
		if (resource == null) {
			// resource is free for use
			resource = m_resources.get(object);
			if (resource == null) {
				Log.log(this, "No resource for " + object.toString());
			}
			grantResource(task, resource);
		} else if (task.getPriority() > getCurrentPriorityCeiling()) {
			// current task is higher priority than all other
			// current tasks on any resource
			grantResource(task, resource);
		} else if (m_activeResources.peek() == resource.getOwner()) {
			// current task is the task creating current priority ceiling
			grantResource(task, resource);
		} else {
			m_taskQueue.promote((ScheduledTask) resource.getOwner(), resource
					.getCeiling());
			new ContextSwitcher(m_taskQueue.peek()).start();
			task.block();
			grantResource(task, resource);
		}
	}

	public static void unlock(Object object) {
		instance.unlockResource(object);
	}

	private void unlockResource(Object object) {
		Log.log(this, "Request to unlock " + object);
		Resource resource = m_activeResources.getResource(object);
		ungrantResource(resource);
	}

	private void grantResource(Task task, Resource resource) {
		resource.setOwner(task);
		m_activeResources.add(resource);
		Log.log(this, resource.toString() + " granted to " + task.toString());
	}

	private void ungrantResource(Resource resource) {
		ScheduledTask owner = (ScheduledTask) resource.getOwner();
		boolean stillPromoted = m_taskQueue.demote(owner);
		if (!stillPromoted) {
			resource.setOwner(null);
		}
		m_activeResources.remove(resource);

		if (m_taskQueue.peek() != owner) {
			new ContextSwitcher(m_taskQueue.peek()).start();
			owner.block();
		}

	}

	private class ScheduledTask implements Task {
		private final Task m_task;
		private final BlockableThread m_thread;
		private boolean m_complete = false;

		ScheduledTask(Task task) {
			m_task = task;
			m_thread = createThread();
		}

		@Override
		public String toString() {
			return m_task.getName();
		}

		public final boolean isComplete() {
			return m_complete;
		}

		public boolean isStarted() {
			return m_thread.isAlive();
		}

		public void block() {
			Log.log(this, "Blocking " + getName());
			synchronized (m_thread) {
				try {
					m_thread.wait();
				} catch (InterruptedException e) {
					// do nothing
				}
			}
		}

		public void release() {
			if (m_thread.isAlive()) {
				synchronized (m_thread) {
					Log.log(this, "Releasing " + getName());
					m_thread.interrupt();
				}
			} else {
				Log.log(this, "Starting " + getName());
				m_thread.setDaemon(false);
				m_thread.start();
			}
		}

		private BlockableThread createThread() {
			BlockableThread thread = new BlockableThread(this);
			thread.setDaemon(false);
			return thread;
		}

		void setComplete() {
			m_complete = true;
		}

		public void doTask() {
			m_task.doTask();
		}

		public long getId() {
			return m_thread.getId();
		}

		public String getName() {
			return m_task.getName();
		}

		public int getOriginalPriority() {
			return m_task.getOriginalPriority();
		}

		public int getPriority() {
			return m_task.getPriority();
		}

		public Collection<Object> getUsedResources() {
			return m_task.getUsedResources();
		}

		public boolean demote() {
			return m_task.demote();
		}

		public void promote(int priority) {
			m_task.promote(priority);
		}
	}

	private class BlockableThread extends Thread {
		private final ScheduledTask m_task;

		public BlockableThread(ScheduledTask task) {
			super(task.getName());
			m_task = task;
			setDaemon(false);
		}

		@Override
		public void run() {
			try {
				m_task.doTask();
			} finally {
				Log.log(this, m_task.getName() + " Complete.");
				m_task.setComplete();
				checkHighestPriorityTask();
			}
		}
	}

	class ContextSwitcher extends Thread {
		private final ScheduledTask m_task;

		ContextSwitcher(final ScheduledTask task) {
			super("ContextSwitcher");
			m_task = task;
		}

		@Override
		public void run() {
			m_task.release();
		}
	}
}
